# -*- coding: utf-8 -*-
from datetime import timedelta, datetime
from time import time

from jms.dm.route.cainiao_new_version.dm_cainiao_route_push_data_dt import jms_dm__dm_cainiao_route_push_data_dt
from utils.operators.rich_sql_sensor import RichSqlSensor
import os
# f = os.popen("hdfs haadmin -getServiceState nn2")
# now = f.read().lower().strip()  # add strip here
#
# if now == "active":
#     namespace = "10.33.48.5:8020"
# else:
#     namespace = "10.33.48.4:8020"

label = f"dwd_cainiao_route_push_data_{int(time())}"
timeout = timedelta(minutes=20).seconds
doris_jms_dm__dm_cainiao_route_push_data_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_cainiao_route_push_data_dt',
    email=['zhangqinglin@jtexpress.com', 'yl_bigdata@yl-scm.com'],
    pool='broker_load_pool',
    conn_id='doris',
    retries=0,
    pre_sql=f"""
             TRUNCATE TABLE jms_dwd.dwd_cainiao_route_push_data PARTITION (p{{{{ execution_date | date_add(1) | cst_ds_nodash }}}});
             LOAD LABEL jms_dwd.{label} (
                 DATA INFILE("hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_cainiao_route_push_data_version2_dt/dt={{{{ execution_date | cst_ds }}}}/*") 
                 INTO TABLE dwd_cainiao_route_push_data FORMAT AS 'PARQUET'
                 (in_from_code,in_from_provider_code,in_from_provider_desc,in_from_city_code,in_from_city_desc,in_from_area_code,in_from_area_desc,from_network_type,in_from_name,reserve_1,from_latest_warehousing,from_departure_days,from_planned_departure,e1_provider_code,e1_provider_desc,e1_city_code,e1_city_desc,e1_area_code,e1_area_desc,e1_network_type,e1_code,e1_name,e1_planned_arrival_day,e1_planned_arrival_time,e1_stop_time,e1_planned_departure_days,e1_planned_departure_time,e2_provider_code,e2_provider_desc,e2_city_code,e2_city_desc,e2_area_code,e2_area_desc,e2_network_type,e2_code,e2_name,e2_planned_arrival_day,e2_planned_arrival_time,e2_stop_time,e2_planned_departure_days,e2_planned_departure_time,e3_provider_code,e3_provider_desc,e3_city_code,e3_city_desc,e3_area_code,e3_area_desc,e3_network_type,e3_code,e3_name,e3_planned_arrival_day,e3_planned_arrival_time,e3_stop_time,e3_planned_departure_days,e3_planned_departure_time,e4_provider_code,e4_provider_desc,e4_city_code,e4_city_desc,e4_area_code,e4_area_desc,e4_network_type,e4_code,e4_name,e4_planned_arrival_day,e4_planned_arrival_time,e4_stop_time,e4_planned_departure_days,e4_planned_departure_time,e5_provider_code,e5_provider_desc,e5_city_code,e5_city_desc,e5_area_code,e5_area_desc,e5_network_type,e5_code,e5_name,e5_planned_arrival_day,e5_planned_arrival_time,e5_stop_time,e5_planned_departure_days,e5_planned_departure_time,e6_provider_code,e6_provider_desc,e6_city_code,e6_city_desc,e6_area_code,e6_area_desc,e6_network_type,e6_code,e6_name,e6_planned_arrival_day,e6_planned_arrival_time,e6_stop_time,e6_planned_departure_days,e6_planned_departure_time,e7_provider_code,e7_provider_desc,e7_city_code,e7_city_desc,e7_area_code,e7_area_desc,e7_network_type,e7_code,e7_name,e7_planned_arrival_day,e7_planned_arrival_time,e7_stop_time,e7_planned_departure_days,e7_planned_departure_time,out_to_provider_code,out_to_provider_desc,out_to_city_code,out_to_city_desc,out_to_area_code,out_to_area_desc,out_network_type,out_to_code,out_to_name,out_planned_arrival_day,out_planned_arrival_time,out_sign_days,out_lastest_time,is_main_route,is_enable,route_id,is_true,latest_warehousing_time_quantum,out_last_shift_code)
                 SET (update_day = DATE('{{{{ execution_date | date_add(1)| cst_ds }}}}'))
              )
             WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
             PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dwd WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dwd WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    # execution_timeout=max(datetime.now().replace(hour=8, minute=25, second=0) - datetime.now(), timedelta(seconds=1)),
)

doris_jms_dm__dm_cainiao_route_push_data_dt << [jms_dm__dm_cainiao_route_push_data_dt ]
